package fun.tan90.easy.binlog.manager.service.impl;

import cn.hutool.core.util.ReUtil;
import cn.hutool.core.util.StrUtil;
import com.mybatisflex.core.FlexGlobalConfig;
import com.mybatisflex.core.datasource.DataSourceKey;
import com.mybatisflex.core.datasource.FlexDataSource;
import com.mybatisflex.core.query.QueryCondition;
import com.mybatisflex.core.row.Db;
import com.mybatisflex.core.row.Row;
import com.mybatisflex.spring.service.impl.ServiceImpl;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import fun.tan90.easy.binlog.base.convention.enums.IErrorCode;
import fun.tan90.easy.binlog.base.convention.exception.ClientException;
import fun.tan90.easy.binlog.manager.dataobject.mapper.BinlogMasterMapper;
import fun.tan90.easy.binlog.manager.dataobject.mapper.BinlogOutputMapper;
import fun.tan90.easy.binlog.manager.dataobject.mapper.BinlogOutputRuleCountMapper;
import fun.tan90.easy.binlog.manager.dataobject.mapper.BinlogOutputRuleMapper;
import fun.tan90.easy.binlog.manager.dataobject.table.BinlogOutputRuleTableDef;
import fun.tan90.easy.binlog.manager.service.BinlogOutputRuleService;
import fun.tan90.easy.binlog.manager.dataobject.BinlogMaster;
import fun.tan90.easy.binlog.manager.dataobject.BinlogOutput;
import fun.tan90.easy.binlog.manager.dataobject.BinlogOutputRule;
import fun.tan90.easy.binlog.manager.dataobject.BinlogOutputRuleCount;
import fun.tan90.easy.binlog.manager.dict.Dict;
import fun.tan90.easy.binlog.manager.model.cmd.BinlogOutputRuleAddCmd;
import fun.tan90.easy.binlog.manager.model.cmd.BinlogOutputRuleUpdateCmd;
import fun.tan90.easy.binlog.manager.model.vo.BinlogOutputRuleVo;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

import static fun.tan90.easy.binlog.manager.dataobject.table.BinlogOutputRuleCountTableDef.BINLOG_OUTPUT_RULE_COUNT;


/**
 * 服务层实现。
 *
 * @author 陈浩杰
 * @since 2023-09-12
 */
@Service
public class BinlogOutputRuleServiceImpl extends ServiceImpl<BinlogOutputRuleMapper, BinlogOutputRule> implements BinlogOutputRuleService {

    private final static List<String> IGNORE_SCHEMAS = Arrays.asList("information_schema", "mysql", "sys", "performance_schema");

    @Resource
    BinlogOutputMapper binlogOutputWayMapper;

    @Resource
    BinlogOutputRuleCountMapper binlogOutputRuleCountMapper;

    @Resource
    BinlogMasterMapper binlogMasterMapper;

    @Override
    public Map<String, List<String>> regularTest(String masterId, String schemaRegular, String tableRegular) {
        BinlogMaster binlogMaster = binlogMasterMapper.selectOneById(masterId);
        if (Objects.isNull(binlogMaster)) {
            throw new ClientException(IErrorCode.BINLOG_1004004);
        }
        String masterCode = binlogMaster.getMasterCode();
        FlexDataSource flexDataSource = FlexGlobalConfig.getDefaultConfig()
                .getDataSource();
        try {
            HikariConfig config = new HikariConfig();
            config.setJdbcUrl(StrUtil.format("jdbc:mysql://{}:{}/{}", binlogMaster.getHostname(), binlogMaster.getPort(), "mysql"));
            config.setUsername(binlogMaster.getUsername());
            config.setPassword(binlogMaster.getPassword());
            flexDataSource.addDataSource(masterCode, new HikariDataSource(config));

            List<Row> rows = DataSourceKey.use(masterCode, () -> Db.selectListBySql("SELECT TABLE_NAME, TABLE_SCHEMA FROM INFORMATION_SCHEMA.TABLES"));
            return rows.stream()
                    .filter(n -> !IGNORE_SCHEMAS.contains(n.getString("TABLE_SCHEMA")))
                    .filter(n -> ReUtil.isMatch(schemaRegular, n.getString("TABLE_SCHEMA")) && ReUtil.isMatch(tableRegular, n.getString("TABLE_NAME")))
                    .collect(Collectors.groupingBy(n -> n.getString("TABLE_SCHEMA"), Collectors.mapping(n -> n.getString("TABLE_NAME"), Collectors.toList())));
        } finally {
            flexDataSource.removeDatasource(masterCode);
        }
    }

    @Override
    public List<BinlogOutputRuleVo> list(String masterId) {
        List<BinlogOutputRule> binlogOutputRules = mapper.selectListByCondition(QueryCondition.create(BinlogOutputRuleTableDef.BINLOG_OUTPUT_RULE.MASTER_ID, masterId));
        return binlogOutputRules.stream().map(n -> {
            String ruleId = n.getId();
            List<BinlogOutputRuleCount> binlogOutputRuleDetails = binlogOutputRuleCountMapper.selectListByCondition(QueryCondition.create(BINLOG_OUTPUT_RULE_COUNT.OUTPUT_RULE_ID, ruleId));
            List<BinlogOutputRuleVo.BinlogMasterOutputRuleDetail> binlogMasterOutputRuleDetails = binlogOutputRuleDetails.stream().map(m -> {
                String outputId = m.getOutputId();
                BinlogOutput binlogOutputWay = binlogOutputWayMapper.selectOneById(outputId);
                return BinlogOutputRuleVo.BinlogMasterOutputRuleDetail
                        .builder()
                        .id(m.getId())
                        .outputWayId(outputId)
                        .outputName(binlogOutputWay.getOutputName())
                        .outputType(binlogOutputWay.getOutputType())
                        .outputTypeCn(Dict.dict(Dict.OUTPUT_TYPE, binlogOutputWay.getOutputType()))
                        .error(m.getError())
                        .countErrorTime(m.getCountErrorTime())
                        .success(m.getSuccess())
                        .countSuccessTime(m.getCountSuccessTime())
                        .createTime(m.getCreateTime())
                        .build();
            }).collect(Collectors.toList());
            return BinlogOutputRuleVo
                    .builder()
                    .ruleId(ruleId)
                    .schemaRegular(n.getSchemaRegular())
                    .tableRegular(n.getTableRegular())
                    .ruleStatus(n.getRuleStatus())
                    .ruleStatusCn(Dict.dict(Dict.RULE_STATUS, n.getRuleStatus()))
                    .details(binlogMasterOutputRuleDetails)
                    .createTime(n.getCreateTime())
                    .build();
        }).collect(Collectors.toList());
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void addBinlogOutputRule(BinlogOutputRuleAddCmd addCmd) {
        String masterId = addCmd.getMasterId();
        BinlogMaster binlogMaster = binlogMasterMapper.selectOneById(masterId);
        if (Objects.isNull(binlogMaster)) {
            throw new ClientException(IErrorCode.BINLOG_1004004);
        }
        String schemaRegular = addCmd.getSchemaRegular();
        String tableRegular = addCmd.getTableRegular();

        long count = mapper.selectCountByCondition(QueryCondition.create(BinlogOutputRuleTableDef.BINLOG_OUTPUT_RULE.MASTER_ID, masterId)
                .and(BinlogOutputRuleTableDef.BINLOG_OUTPUT_RULE.SCHEMA_REGULAR.eq(schemaRegular))
                .and(BinlogOutputRuleTableDef.BINLOG_OUTPUT_RULE.TABLE_REGULAR.eq(tableRegular)));
        if (count > 0) {
            throw new ClientException(IErrorCode.BINLOG_1004005);
        }
        BinlogOutputRule binlogOutputRule = BinlogOutputRule.builder()
                .masterId(masterId)
                .schemaRegular(schemaRegular)
                .tableRegular(tableRegular)
                .ruleStatus(addCmd.getRuleStatus())
                .build();
        mapper.insert(binlogOutputRule);
        String outputRuleId = binlogOutputRule.getId();

        binlogOutputRuleCountMapper.insertBatch(addCmd.getOutputIds()
                .stream()
                .map(outputId -> BinlogOutputRuleCount.builder().outputRuleId(outputRuleId).outputId(outputId).success(0).error(0).build()).collect(Collectors.toList())
        );
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateBinlogOutputRule(BinlogOutputRuleUpdateCmd updateCmd) {
        String id = updateCmd.getId();
        mapper.update(BinlogOutputRule.builder()
                .ruleStatus(updateCmd.getRuleStatus())
                .schemaRegular(updateCmd.getSchemaRegular())
                .tableRegular(updateCmd.getTableRegular())
                .id(id).build());
        Set<String> outputWayIds = updateCmd.getOutputWayIds();
        List<BinlogOutputRuleCount> binlogOutputRuleDetails = binlogOutputRuleCountMapper.selectListByCondition(QueryCondition.create(BINLOG_OUTPUT_RULE_COUNT.OUTPUT_RULE_ID, id));
        Set<String> dbOutputWayIds = binlogOutputRuleDetails.stream().map(BinlogOutputRuleCount::getOutputId).collect(Collectors.toSet());

        List<String> delOutputWayIds = dbOutputWayIds.stream().filter(n -> !outputWayIds.contains(n)).collect(Collectors.toList());
        if (!delOutputWayIds.isEmpty()) {
            binlogOutputRuleCountMapper.deleteByCondition(QueryCondition.create(BINLOG_OUTPUT_RULE_COUNT.OUTPUT_RULE_ID, id).and(BINLOG_OUTPUT_RULE_COUNT.OUTPUT_ID.in(delOutputWayIds)));
        }

        List<String> addOutputWayIds = outputWayIds.stream().filter(n -> !dbOutputWayIds.contains(n)).collect(Collectors.toList());
        if (!addOutputWayIds.isEmpty()) {
            binlogOutputRuleCountMapper.insertBatch(addOutputWayIds
                    .stream()
                    .map(outputWayId -> BinlogOutputRuleCount.builder().outputRuleId(id).outputId(outputWayId).success(0).error(0).build()).collect(Collectors.toList())
            );
        }
    }

    @Override
    public void delBinlogOutputRule(String id) {
        binlogOutputRuleCountMapper.deleteByCondition(QueryCondition.create(BINLOG_OUTPUT_RULE_COUNT.OUTPUT_RULE_ID, id));
        mapper.deleteById(id);
    }

    @Override
    public void switchBinlogOutputRuleStatus(String id) {
        BinlogOutputRule binlogOutputRule = mapper.selectOneById(id);
        String ruleStatus = binlogOutputRule.getRuleStatus();
        ruleStatus = "0".equals(ruleStatus) ? "1" : "0";
        mapper.update(BinlogOutputRule.builder().ruleStatus(ruleStatus).id(id).build());
    }
}
